﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Windows;
using System.Windows.Controls;
using System.Windows.Data;
using System.Windows.Documents;
using System.Windows.Input;
using System.Windows.Media;
using System.Windows.Media.Imaging;
using System.Windows.Navigation;
using System.Windows.Shapes;
using RabbitMQ.Client;

namespace RabbitMQ示例
{
    /// <summary>
    /// 带应答的MQ应用.xaml 的交互逻辑
    /// </summary>
    public partial class 带应答的MQ应用 : UserControl
    {
        public 带应答的MQ应用()
        {
            InitializeComponent();
        }

        /// <summary>
        /// mq创建对像
        /// </summary>
        RabbitMQ.Client.ConnectionFactory faction = new RabbitMQ.Client.ConnectionFactory();

        /// <summary>
        /// 创建联接
        /// </summary>
        RabbitMQ.Client.IConnection conn = null;

        /// <summary>
        /// 初始化配置
        /// </summary>
        MqConfig conf = null;
        /// <summary>
        /// 联接模式字典
        /// </summary>
        private System.Collections.Concurrent.ConcurrentDictionary<string, MQChannelEntity> channelDic = new System.Collections.Concurrent.ConcurrentDictionary<string, MQChannelEntity>();
        /// <summary>
        /// 联接模式字典
        /// </summary>
        public ConcurrentDictionary<string, MQChannelEntity> ChannelDic
        {
            get
            {
                return channelDic;
            }

        }

        /// <summary>
        /// 初始化工厂对像
        /// </summary>
        public void Init()
        {
            //配置信息
            MqConfig conf = new MqConfig()
            {
                HostName = "192.168.2.110",
                UserName = "rabbitmq",
                Password = "rabbitmq",
                VirtualHost = "/",
                QueueCollection = new List<MQQueueInfo>() {
                     new MQQueueInfo() {  NeedAck=true,
                         DeliveryMode =2,
                         ExchangeName = "IN.Test",
                         QueueName ="IN.RelQue",
                         ExchangeType = RabbitMQ.Client.ExchangeType.Fanout,
                          RoutingKey="IN.RelQueue#",
                           Durable = true,

                     }

                },
            };

            System.Xml.Serialization.XmlSerializer xml = new System.Xml.Serialization.XmlSerializer(typeof(MqConfig));
            using (System.IO.StringWriter sw = new System.IO.StringWriter())
            {
                xml.Serialize(sw, conf);
                var str = sw.ToString();
            }


            faction.HostName = conf.HostName;
            faction.UserName = conf.UserName;
            faction.Password = conf.Password;
            faction.VirtualHost = conf.VirtualHost;
            //集群56720
            faction.Port = 56720;
            //设置自动恢复功能
            faction.AutomaticRecoveryEnabled = true;
            faction.NetworkRecoveryInterval = new TimeSpan(0, 0, 5);
            faction.ContinuationTimeout = new TimeSpan(0, 0, 5);
            //创建联接对像
            conn = faction.CreateConnection(conf.HostName.Split(','));
            foreach (var item in conf.QueueCollection)
            {
                //创建或者设置频道
                RabbitMQ.Client.IModel channel = conn.CreateModel();
                //设置Exchange持久化
                channel.ExchangeDeclare(item.ExchangeName, item.ExchangeType, item.Durable);
                //创建队列 ,非排它,非自动删除
                channel.QueueDeclare(item.QueueName, item.Durable, false, false, null);
                channel.QueueBind(item.QueueName, item.ExchangeName, item.RoutingKey, null);
                //添加到集合中
                ChannelDic.TryAdd(item.RoutingKey, new MQChannelEntity() { Channel = channel, QueueInfo = item });
            }





        }

        long ina = 0;

        /// <summary>
        /// 启用应答机制处理
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void button_Click(object sender, RoutedEventArgs e)
        {

            MQChannelEntity channel;
            if (ChannelDic.TryGetValue(ChannelDic.Keys.First(), out channel))
            {

                if (channel.Channel.IsOpen)
                {
                    System.Threading.Interlocked.Increment(ref ina);
                    string message = "Hello World " + Guid.NewGuid() + " " + ina;
                    var body = Encoding.UTF8.GetBytes(message);

                    //发布数据
                    channel.Channel.BasicPublish(channel.QueueInfo.ExchangeName, channel.QueueInfo.RoutingKey, null, body);
                }
            }

        }
        /// <summary>
        /// 步骤1初始化
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void button1_Click(object sender, RoutedEventArgs e)
        {

            Init();

        }

        public void 启用应答机制()
        {
            MQChannelEntity channel;
            if (ChannelDic.TryGetValue(ChannelDic.Keys.First(), out channel))
            {

                if (channel.Channel.IsOpen)
                {
                    try { channel.Channel.BasicQos(0, 20, false); }
                    catch (Exception ex)
                    {
                        Console.WriteLine(ex);
                    }

                    RabbitMQ.Client.QueueingBasicConsumer queue = new QueueingBasicConsumer(channel.Channel);
                    // var queuea = channel.Channel.QueueDeclare();
                    channel.Channel.BasicConsume(channel.QueueInfo.QueueName, true, queue);

                    long n = 0;
                    //丢失的数据项
                    List<long> listLost = new List<long>();

                    System.Threading.Tasks.Task.Factory.StartNew(() =>
                    {
                        while (queue.IsRunning || channel.Channel.IsOpen)
                        {
                            ReadData(queue,n);
                        }
                    });


                    System.Threading.Tasks.Task.Factory.StartNew(() =>
                    {
                        while (queue.IsRunning || channel.Channel.IsOpen)
                        {
                            Console.WriteLine("已经收到了{0}条信息", n);
                            System.Threading.Thread.Sleep(1000 * 10);
                        }
                    });

                }
            }
        }


        public static void ReadData(RabbitMQ.Client.QueueingBasicConsumer queue,long n)
        {
            var q = queue.Queue.Dequeue();
            if (q != null)
            {
                //q.Redelivered = true;
                //channel.Channel.BasicAck(q.DeliveryTag, false);
                // var k = ProtoBufDerialize(q.Body);
                System.Threading.Interlocked.Increment(ref n);
                //string str = System.Text.UTF8Encoding.UTF8.GetString(q.Body);
                //Console.WriteLine(str);
            }
        }

        /// <summary>
        /// 将一个对像序列化为Protobuf的二进制数据
        /// </summary>
        /// <param name="obj">数据实体对像</param>
        /// <returns>返回数据,如果</returns>
        public byte[] ProtoBufSerialize(DIS.DataBus.Models.RealDataMessages.RealTimeDataMsg obj)
        {
            byte[] result = null;
            using (System.IO.MemoryStream stream = new System.IO.MemoryStream())
            {
                //序列化数据
                ProtoBuf.Serializer.Serialize(stream, obj);
                result = stream.ToArray();
                stream.Close();
            }
            return result;
        }
        /// <summary>
        /// 将一个二进制数据对像序列化为Protobuf
        /// </summary>
        /// <param name="obj">数据实体对像</param>
        /// <returns>返回数据,如果</returns>
        public DIS.DataBus.Models.RealDataMessages.RealTimeDataMsg ProtoBufDerialize(byte[] obj)
        {
            DIS.DataBus.Models.RealDataMessages.RealTimeDataMsg result = null;
            using (System.IO.MemoryStream stream = new System.IO.MemoryStream(obj))
            {
                //序列化数据
                result = ProtoBuf.Serializer.Deserialize<DIS.DataBus.Models.RealDataMessages.RealTimeDataMsg>(stream);
                stream.Close();
            }
            return result;
        }
        private void button2_Click(object sender, RoutedEventArgs e)
        {
            启用应答机制();
        }

        MQChannelEntity kinit()
        {
            var conf = RabbitMQConfig.Instance.Conf;
            var factory = new ConnectionFactory();
            factory.HostName = conf.HostName;
            factory.UserName = conf.UserName;
            factory.Password = conf.Password;
            factory.VirtualHost = conf.VirtualHost;
            //设置端口
            if (conf.Port > 0) {
                factory.Port = conf. Port;
            }
            //集群端口
            //faction.Port = 56720;
            //设置自动恢复功能
            factory.AutomaticRecoveryEnabled = true;
            factory.NetworkRecoveryInterval = new TimeSpan(0, 0, 5);
            factory.ContinuationTimeout = new TimeSpan(0, 0, 5);
            //创建联接对像
            conn = factory.CreateConnection(conf.HostName.Split(','));


            var item = RabbitMQConfig.Instance.Conf.QueueCollection[1];
            //创建或者设置频道
            RabbitMQ.Client.IModel channel = conn.CreateModel();
            //设置Exchange持久化
            channel.ExchangeDeclare(item.ExchangeName, item.ExchangeType, item.Durable);
            //创建队列 ,非排它,非自动删除
            channel.QueueDeclare(item.QueueName, item.Durable, false, false, null);
            channel.QueueBind(item.QueueName, item.ExchangeName, item.RoutingKey, null);

            return new MQChannelEntity() { Channel = channel, QueueInfo = item, factory = factory };
        }



        MQChannelEntity cdddd = null;

        /// <summary>
        /// 新启用一个写入线程factory
        /// 
        /// 1.一个channel多线程发布不能提高发布性能
        /// 2.
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void button3_Click(object sender, RoutedEventArgs e)
        {

#if 一个channel多线程写不能提高发布性能
            if (cdddd == null)
            {
                //初始化
                cdddd = kinit();
            }
            #region 创建多个factory
            System.Threading.Tasks.Task.Factory.StartNew(() =>
            {

                string xxx = System.IO.File.ReadAllText(AppDomain.CurrentDomain.BaseDirectory + "/inifiles/示例数据.txt");
                byte[] arr = Convert.FromBase64String(xxx.Replace("\r\n", ""));
                //发布数据
                while (true)
                {
                    cdddd.Channel.BasicPublish(cdddd.QueueInfo.ExchangeName, string.Empty, null, arr);
                }
            });



            #endregion
#endif

            #region 一个factory多个conn,每个conn一个Channel,使用不同的connel发布可以提高发布性能

            if (cdddd == null)
            {
                //初始化
                cdddd = kinit();
            }

            conn = cdddd.factory.CreateConnection();
            var item = RabbitMQConfig.Instance.Conf.QueueCollection[1];
            //创建或者设置频道
            RabbitMQ.Client.IModel channel = conn.CreateModel();
            ////设置Exchange持久化
            channel.ExchangeDeclare(item.ExchangeName, item.ExchangeType, item.Durable);
            //创建队列 ,非排它,非自动删除
            channel.QueueDeclare(item.QueueName, item.Durable, false, false, null);
            channel.QueueBind(item.QueueName, item.ExchangeName, item.RoutingKey, null);


            ////////////////////////////////////////////////
            System.Threading.Thread thr = new System.Threading.Thread(() =>
            {
                string xxx = System.IO.File.ReadAllText(AppDomain.CurrentDomain.BaseDirectory + "/inifiles/示例数据.txt");
                byte[] arr = Convert.FromBase64String(xxx.Replace("\r\n", ""));
                //byte[] arr1 = arr.Take(5000).ToArray();
                //发布数据
                for (int i = 0; i < 99999999; i++)
                {
                    var s =string.Format(item.RoutingKey, i);
                    channel.BasicPublish(cdddd.QueueInfo.ExchangeName, s, null, arr);
                }

            });
            thr.Priority = System.Threading.ThreadPriority.Highest;
            thr.Start();

            System.Threading.Tasks.Task.Factory.StartNew(() =>
            {

            });
            #endregion


        }
    }

    #region 内部使用类
    /// <summary>
    /// MQ对像实例
    /// </summary>
    public class MQChannelEntity
    {

        public IModel Channel { get; set; }

        /// <summary>
        /// 邦定的队列信息
        /// </summary>
        public MQQueueInfo QueueInfo { get; set; }
        /// <summary>
        /// 联接创建工厂
        /// </summary>
        public ConnectionFactory factory { get; set; }
    }
    #endregion



}
/// <summary>
/// MQ透传实体
/// </summary>
[ProtoBuf.ProtoContract]
public class CmdMQTransparent
{
    /// <summary>
    /// 协议类型如:x68
    /// </summary>
    [ProtoBuf.ProtoMember(1)]
    public string PType { get; set; }

    /// <summary>
    /// 用户标识,用于用户传输时自己填充
    /// </summary>
    [ProtoBuf.ProtoMember(2)]
    public string UserToKen { get; set; }


    /// <summary>
    /// 终端号,内部识别用
    /// </summary>
    [ProtoBuf.ProtoMember(3)]
    public long TerminalCode { get; set; }
    /// <summary>
    /// VIN码或者标识码
    /// </summary>
    [ProtoBuf.ProtoMember(4)]
    public string VIN { get; set; }

    /// <summary>
    /// 传输数据内容 
    ///  key ,用x68,x0A,0x3456 表示 协议号,协议消息ID
    ///  value,用序列化的字符串表示可以是protobuf的Base64字符串也可以是JSOn,XML字符串
    /// </summary>
    [ProtoBuf.ProtoMember(5)]
    public Dictionary<string, string> Content { get; set; }

    /// <summary>
    ///  扩展字段Key为String 值为 string
    /// </summary>
    [ProtoBuf.ProtoMember(9990)]
    public Dictionary<string, string> EXT { get; set; }
    /// <summary>
    /// 扩展字段Key为String 值为 Byte[]
    /// </summary>
    [ProtoBuf.ProtoMember(9991)]
    public Dictionary<string, byte[]> EXTA { get; set; }
    /// <summary>
    /// 扩展字段Key为String 值为 double
    /// </summary>
    [ProtoBuf.ProtoMember(9992)]
    public Dictionary<string, double> EXTB { get; set; }


}


